// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/core/api.h>
#include <nebula/compute/api.h>
#include <nebula/csv/api.h>
#include <nebula/csv/writer.h>
#include <nebula/io/api.h>
#include <nebula/ipc/json_simple.h>
#include <turbo/utility/status.h>

#include <iostream>
#include <vector>

using DatumVector = std::vector<nebula::Datum>;


// Many operations in Apache Nebula operate on
// columns of data, and the columns of data are
// assembled into a table. In this example, we
// examine how to compare two arrays which are
// combined to form a table that is then written
// out to a CSV file.
//
// To run this example you can use
// ./compute_and_write_csv_example
//
// the program will write the files into
// compute_and_write_output.csv
// in the current directory

turbo::Status run_main(int argc, char **argv) {
    // Make Arrays
    nebula::StringBuilder str_builder;
    nebula::StringBuilder sum_builder;

    // Make place for 8 values in total
    TURBO_RETURN_NOT_OK(str_builder.resize(8));
    TURBO_RETURN_NOT_OK(sum_builder.resize(8));

    // Bulk append the given values
    std::vector<std::string> string_values = {"a", "2", "3", "4", "5", "6", "7", "8"};
    DatumVector inputs;
    TURBO_RETURN_NOT_OK(str_builder.append_values(string_values));
    std::shared_ptr<nebula::StringArray> array_a;
    TURBO_RETURN_NOT_OK(str_builder.finish(&array_a));
    inputs.push_back(array_a);
    str_builder.reset();
    string_values = {"2", "5", "1", "3", "6", "2", "7", "4"};
    std::shared_ptr<nebula::StringArray> array_b;
    TURBO_RETURN_NOT_OK(str_builder.append_values(string_values));
    TURBO_RETURN_NOT_OK(str_builder.finish(&array_b));

    string_values = {"aa", "bb", "cc", "dd", "ee", "ff", "gg", "hh"};
    std::shared_ptr<nebula::StringArray> array_d;
    TURBO_RETURN_NOT_OK(str_builder.append_values(string_values));
    TURBO_RETURN_NOT_OK(str_builder.finish(&array_d));

    std::shared_ptr<nebula::StringArray> array_c;
    string_values = {""};
    TURBO_RETURN_NOT_OK(str_builder.append_values(string_values));
    TURBO_RETURN_NOT_OK(str_builder.finish(&array_c));
    std::shared_ptr<nebula::Scalar> sep = std::move(array_c->get_scalar(0)).value_or_die();
    // Cast the arrays to their actual types
    auto str_array_a = std::static_pointer_cast<nebula::StringArray>(array_a);
    auto str_array_b = std::static_pointer_cast<nebula::StringArray>(array_b);
    auto str_array_d = std::static_pointer_cast<nebula::StringArray>(array_d);
    // Explicit comparison of values using a loop
    KLOG(INFO)<<"str_array_a length:"<<str_array_a->length();
    KLOG(INFO)<<"str_array_b length:"<<str_array_b->length();
    nebula::StringBuilder r_builder;
    for (int64_t i = 0; i < 8; i++) {
        KLOG(INFO)<<"loop:"<<i;
        if ((!str_array_a->is_null(i)) && (!str_array_b->is_null(i))) {
            KLOG(INFO)<<"a:"<<str_array_a->value(i)<<" b:"<<str_array_b->value(i);
           auto joined = turbo::str_cat(str_array_a->value(i), str_array_b->value(i),  str_array_d->value(i));
            KLOG(INFO)<<"joined:"<<joined;
            r_builder.append(joined).abort();
        } else {
            KLOG(INFO)<<"loop:"<<i<<" null";
            r_builder.append_null().abort();
        }
    }

    string_values = {"12", "25", "31", "43", "56", "62", "77", "84"};
    //TURBO_RETURN_NOT_OK(sum_builder.append_values(string_values));
    std::shared_ptr<nebula::StringArray> array_a_gt_b_self;
    TURBO_RETURN_NOT_OK(r_builder.finish(&array_a_gt_b_self));
    std::cout << "Array explicitly compared" << std::endl;

    // Explicit comparison of values using a compute function
    inputs.emplace_back(array_b);
    inputs.emplace_back(array_d);
    inputs.emplace_back(sep);
    TURBO_MOVE_OR_RAISE(nebula::Datum compared_datum,
            nebula::compute::call_function("binary_join_element_wise", inputs));
    auto array_a_gt_b_compute = compared_datum.make_array();
    std::cout << "Arrays compared using a compute function" << std::endl;
    std::cout << array_a_gt_b_compute->to_string() << std::endl;
    // create a table for the output
    auto schema =
            nebula::schema({nebula::field("a", nebula::utf8()), nebula::field("b", nebula::utf8()),
                            nebula::field("a+b (self written)", nebula::utf8()),
                            nebula::field("a+b (nebula)", nebula::utf8())});
    std::shared_ptr<nebula::Table> my_table = nebula::Table::create(
            schema, {array_a, array_b, array_a_gt_b_self, array_a_gt_b_compute});

    std::cout << "Table created" << std::endl;

    // write table to CSV file
    auto csv_filename = "compute_string_add_output.csv";
    TURBO_MOVE_OR_RAISE(auto outstream, nebula::io::FileOutputStream::open(csv_filename));

    std::cout << "Writing CSV file" << std::endl;
    TURBO_RETURN_NOT_OK(nebula::csv::write_csv(
            *my_table, nebula::csv::WriteOptions::defaults(), outstream.get()));

    return turbo::OkStatus();
}

int main(int argc, char **argv) {
    turbo::Status status = run_main(argc, argv);
    if (!status.ok()) {
        std::cerr << status.to_string() << std::endl;
        return EXIT_FAILURE;
    }
    return EXIT_SUCCESS;
}
